package penguin.transfer.netty.common;

import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.buffer.ChannelBuffers;
import org.jboss.netty.channel.*;
import org.jboss.netty.channel.ChannelHandler.Sharable;
import org.jboss.netty.handler.codec.frame.FrameDecoder;
import org.jboss.netty.handler.codec.oneone.OneToOneEncoder;
import penguin.serialize.SerializeClient;
import penguin.serialize.hessian.HessianClient;


/**
 * NettyCodecAdapter.
 * 
 */
public class NettyCodecAdapter {

    private final ChannelHandler encoder = new InternalEncoder();
    private final ChannelHandler decoder = new InternalDecoder();

    private final int HEADER_LENGTH = 4;

    private final SerializeClient serializeClient = new HessianClient();

    public NettyCodecAdapter() {
    }

    public ChannelHandler getEncoder() {
        return encoder;
    }

    public ChannelHandler getDecoder() {
        return decoder;
    }


    @Sharable
    private class InternalEncoder extends OneToOneEncoder {

        @Override
        protected Object encode(ChannelHandlerContext ctx, Channel ch, Object msg) throws Exception {

              byte[] bytes = serializeClient.serialize(msg);

              int length = bytes.length;

              ChannelBuffer channelBuffer = ChannelBuffers.dynamicBuffer();
              channelBuffer.writeInt(length);
              channelBuffer.writeBytes(bytes);

            return channelBuffer;
        }
    }

    private class InternalDecoder extends FrameDecoder {

        @Override
        protected Object decode(ChannelHandlerContext context, Channel channel,ChannelBuffer buffer) throws Exception {

            int readable = buffer.readableBytes();
            if (readable < HEADER_LENGTH) {
                return null;
            }

            buffer.markReaderIndex();

            int length = buffer.readInt();

            if(buffer.readableBytes() < length){
                buffer.resetReaderIndex();
                return null;
            }

            ChannelBuffer channelBuffer = buffer.readBytes(length);

            Object object = serializeClient.deserialize(channelBuffer.array());

           return object;
        }

    }
}